package com.sf.kafka;

import java.util.List;
import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class KafkaProducerClient<K,V> {
	public Producer<K, V> producer;
	Properties props = new Properties();
	
	public KafkaProducerClient() {
		props.setProperty("bootstrap.servers", "localhost:9092");
		props.setProperty("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
		props.setProperty("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
		//如果这里设置0不确认，则属性retries不会生效
		props.setProperty("acks", "1");
		//重试次数
		props.setProperty("retries", String.valueOf(Integer.MAX_VALUE));
		//当发送端发送速度跟不上时，缓存在内存的大小，默认32M
		props.setProperty("buffer.memory", "33554432");
		//消息缓存在buffer.memory的时间，如果缓存时间超过默认值10S，会抛异常
		props.setProperty("max.block.ms", "60000");
		//设置为true时，MAX_BLOCK_MS属性会变成Long.MAX_VALUE，也就是永远不会抛出异常
		props.setProperty("block.on.buffer.full", "true");
		props.setProperty("compression.type", "snappy");
		props.setProperty("producer.type", "async");
		//kafka producer批次发送batch.size的大小
		props.setProperty("batch.size", "16384");
		producer = new KafkaProducer<>(props);
	}
	
	public void send(SFMessage<K,V> message) {
		producer.send(new ProducerRecord<K, V>(message.getTopicName(), message.getKey(), message.getValue()));
	}
	
	public void send(List<SFMessage<K,V>> messageList) {
		for(int i=0;i<messageList.size();i++) {
			producer.send(new ProducerRecord<K, V>(messageList.get(i).getTopicName(), messageList.get(i).getKey(), messageList.get(i).getValue()));
		}
		producer.flush();
	}
	
	public void close() {
		producer.flush();
		producer.close();
	}
	
	public class SFMessage<K,V>{
		private String topicName;
		private K key;
		private V value;
		
		public SFMessage(String topicName, K key, V value) {
			this.topicName = topicName;
			this.key = key;
			this.value = value;
		}
		public String getTopicName() {
			return topicName;
		}
		public K getKey() {
			return key;
		}
		public V getValue() {
			return value;
		}
	}
	
	public static void main(String[] args) {
		KafkaProducerClient client = new KafkaProducerClient<byte[],byte[]>();
			byte[] key = String.valueOf(System.currentTimeMillis()).getBytes();
			byte[] value = ("hello world,hello world").getBytes();
			KafkaProducerClient.SFMessage message = client.new SFMessage("last-line-test", key, value);
			client.send(message);
		try {
			Thread.sleep(10000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		client.close();
	}
}
